#include "msg_queue.h"
#include <assert.h>
#include <pthread.h>

#define SPIN_INIT(q) spinlock_init(&(q)->lock);
#define SPIN_LOCK(q) spinlock_lock(&(q)->lock);
#define SPIN_UNLOCK(q) spinlock_unlock(&(q)->lock);
#define SPIN_DESTROY(q) spinlock_destroy(&(q)->lock);

static inline void
spinlock_init(struct spinlock *lock) {
	lock->lock = 0;
}

static inline void
spinlock_lock(struct spinlock *lock) {
	while (__sync_lock_test_and_set(&lock->lock,1)) {}
}

// static inline int
// spinlock_trylock(struct spinlock *lock) {
//	 return __sync_lock_test_and_set(&lock->lock,1) == 0;
// }

static inline void
spinlock_unlock(struct spinlock *lock) {
	__sync_lock_release(&lock->lock);
}

//static inline void
//spinlock_destroy(struct spinlock *lock) {
// 	(void) lock;
//}

struct global_queue {
	struct message_queue *head;
	struct message_queue *tail;
	struct spinlock lock;
	pthread_cond_t cond;
	pthread_mutex_t mutex;
	int quit;	
};

static struct global_queue *Q;

void skynet_globalmq_init() {
	static struct global_queue __Q;
	Q = &__Q;
	SPIN_INIT(Q);
	Q->head = Q->tail = NULL;	
	pthread_mutex_init(&Q->mutex, NULL);
	pthread_cond_init(&Q->cond, NULL);
	Q->quit = 0;
}

static void globalmq_wakeup() {
	pthread_cond_signal(&Q->cond);
}

int globalmq_sleep() {
	if (pthread_mutex_lock(&Q->mutex) == 0)
	{
		//				++ m->sleep;
		// "spurious wakeup" is harmless,
		// because skynet_context_message_dispatch() can be call at any time.
		if (!Q->quit)
			pthread_cond_wait(&Q->cond, &Q->mutex);
		//				-- m->sleep;
		if (pthread_mutex_unlock(&Q->mutex))
		{
			fprintf(stderr, "unlock mutex error");
			exit(1);
		}
	}
	return (0);
}

void 
skynet_globalmq_push(struct message_queue * queue) {
	struct global_queue *q= Q;

	SPIN_LOCK(q);
	assert(queue->next == NULL);
	assert(q->tail != queue);	
	if(q->tail) {
		q->tail->next = queue;
		q->tail = queue;
	} else {
		q->head = q->tail = queue;
	}
	SPIN_UNLOCK(q);
	globalmq_wakeup();
}

struct message_queue * 
skynet_globalmq_pop() {
	struct global_queue *q = Q;

	SPIN_LOCK(q)
	struct message_queue *mq = q->head;
	if(mq) {
		q->head = mq->next;
		if(q->head == NULL) {
			assert(mq == q->tail);
			q->tail = NULL;
		}
		mq->next = NULL;
	}
	SPIN_UNLOCK(q)

	return mq;
}

void normal_queue_init(struct normal_queue *q) {
	SPIN_INIT(q);
	q->head = q->tail = NULL;
}
void normal_queue_push(struct normal_queue *q, struct message_queue * queue) {
	SPIN_LOCK(q);
	assert(queue->next == NULL);
	assert(q->tail != queue);
	if(q->tail) {
		q->tail->next = queue;
		q->tail = queue;
	} else {
		q->head = q->tail = queue;
	}
	SPIN_UNLOCK(q);
}

struct message_queue *normal_queue_pop(struct normal_queue *q) {
	SPIN_LOCK(q)
	struct message_queue *mq = q->head;
	if(mq) {
		q->head = mq->next;
		if(q->head == NULL) {
			assert(mq == q->tail);
			q->tail = NULL;
		}
		mq->next = NULL;
	}
	SPIN_UNLOCK(q)
	return mq;
}

struct message_queue *normal_queue_getmq(struct normal_queue *q) {
	SPIN_LOCK(q)
	struct message_queue *mq = q->head;
	q->head = q->tail = NULL;
	SPIN_UNLOCK(q)
	return mq;
}

bool is_normal_queue_empty(struct normal_queue *q)
{
	bool ret = true;
	SPIN_LOCK(q)
	if (q->head)
		ret = false;
	SPIN_UNLOCK(q)
	return ret;
}
